package sjc;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * Barrier是一个同步点，每一个进程到达此点都要等待，直到某一个条件满足，然后所有的节点继续进行。
 *
 * JDK的并发包下有CyclicBarrier，它看起来和CountDownLatch有很大的相似之处：
 *      - CountDownLatch：是一个同步的辅助类，允许一个或多个线程，等待其他一组线程完成操作，再继续执行。CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。
 *      - CyclicBarrier: 是一个同步的辅助类，允许一组线程相互之间等待，达到一个共同点，再继续执行。
 *      - 我们需要 10名参赛者,每当有一名人报名,需要的人数就减一,直到报满10个人为止。这个时候就用CountDownLatch，
 *      - 假如说下午2点开始比赛,有的参赛者来的早,那么它需要等待其他参赛者到来之后才开始进行比赛,这个时候就用 CyclicBarrier。
 *
 *  如果是在同一个JVM中的话，使用CyclicBarrier完全可以解决诸如此类的多线程同步问题。但是，如果在分布式环境中又该如何解决呢？
 *  Curator中提供DistributedBarrier就是用来实现分布式Barrier的。很多台机器都创建了一个barrier，此时他们都不能动了，除非满足一个条件，才能让他们一起往下动。
 */
public class DistributedBarrierTest {
    private static final String PATH = "/examples/barrier";
    private static DistributedBarrier barrier;

    public static void main(String[] args) throws Exception {
        for (int i=0; i < 5; i++){
            final int index = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try{
                        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3, Integer.MAX_VALUE);
                        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.72.133:2181", 3000, 3000, retryPolicy);
                        client.start();

                        barrier = new DistributedBarrier(client, PATH);
                        System.out.println("线程 "+index+" 等待");
                        barrier.setBarrier();
                        barrier.waitOnBarrier();
                        System.out.println("线程 "+index+" 已经执行");
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }).start();
        }

        Thread.sleep(5 * 1000);

        if(barrier != null){
            System.out.println("所有线程已到达，准备启动");
            barrier.removeBarrier();
        }
    }

}
